package callback

import (
	"encoding/gob"
	"fmt"
	"log"
	"os"
	"sync"
	"time"
	"xin-job/client/cbiz"
	"xin-job/client/sconf"
	"xin-job/core/biz"
)

var instance = &TriggerCallback{
	q:    make([]*biz.HandleCallbackParam, 0),
	size: 0,
}

func GetInstance() *TriggerCallback {
	return instance
}

type TriggerCallback struct {
	q    []*biz.HandleCallbackParam // 利用切片实现简单队列
	size int
	sync.Mutex
	stopFlag chan struct{}
}

func (t *TriggerCallback) Start() {
	go t.callback()
	go t.retryCallback()
}

func (t *TriggerCallback) Stop() {
	t.stopFlag <- struct{}{}
}

func (t *TriggerCallback) callback() {
	defer func() {
		if err := recover(); err != nil {
			log.Printf("trigger callback err %v\n", err)
		}
	}()

	for {
		select {
		case <-t.stopFlag:
			// TODO 优雅停机
		default:
			if t.size > 0 {
				data := t.drainTo()
				t.doCallback(data)
			}
		}
	}
}

func (t *TriggerCallback) retryCallback() {
	defer func() {
		if err := recover(); err != nil {
			log.Printf("trigger retry callback err %v\n", err)
		}
	}()

	for {
		select {
		case <-t.stopFlag:
		default:
			t.retryFailCallbackFile()
			time.Sleep(30 * time.Second)
		}

	}
}

func (t *TriggerCallback) PushCallBack(param *biz.HandleCallbackParam) {
	t.Mutex.Lock()
	defer t.Mutex.Unlock()
	t.q = append(t.q, param)
	t.size++
}

// drainTo 队列中取出数据
func (t *TriggerCallback) drainTo() []*biz.HandleCallbackParam {
	if t.size == 0 {
		return nil
	}

	t.Mutex.Lock()
	defer t.Mutex.Unlock()

	n := make([]*biz.HandleCallbackParam, 0)
	n = append(n, t.q...)
	t.q = make([]*biz.HandleCallbackParam, 0)

	return n
}

// doCallback 回调
func (t *TriggerCallback) doCallback(params []*biz.HandleCallbackParam) {
	callbackRet := false
	if len(params) == 0 {
		return
	}
	for _, address := range sconf.ServerConf.AdminAddressList {
		client := cbiz.GetBizClient()
		client.AdminAddressUrl = address
		result := client.Callback(params)
		if result.Code == 200 {
			// 回调成功
			callbackRet = true
			log.Printf(">>>>>>>>>>> xxl-job callback success, params:%v, \n", params)
			break // 调度中心可能为集群，回调到其中一台成功即可
		} else {
			// 记录失败日志
			// TODO 本地日志
		}
	}

	if !callbackRet {
		// 回调失败，将 params 写入本地文件
		t.appendFailCallbackFile(params)
	}
}

// 把回调失败的数据存储到本地一个专门的文件当中
func (t *TriggerCallback) appendFailCallbackFile(params []*biz.HandleCallbackParam) {
	// TODO 暂时写入当前目录，后续需要改为配置文件
	pwd, _ := os.Getwd() // 当前目录

	logPath := fmt.Sprintf("%s/callbacklog", pwd)

	_, err := os.Stat(logPath)
	if !os.IsExist(err) {
		os.MkdirAll(logPath, 0775)
	} else {
		log.Printf("appendFailCallbackFile dir err=%s", err.Error())
	}

	logFile := fmt.Sprintf("%s/xxl-job-callback-%d.log", logPath, time.Now().UnixMilli())

	file, err := os.Create(logFile)
	if err != nil {
		log.Printf("appendFailCallbackFile err=%s", err.Error())
	}
	defer file.Close()

	// gob 序列化
	e := gob.NewEncoder(file)

	err = e.Encode(params)
	if err != nil {
		log.Printf("appendFailCallbackFile err=%s", err.Error())
	}
}

func (t *TriggerCallback) retryFailCallbackFile() {
	pwd, _ := os.Getwd()

	logPath := fmt.Sprintf("%s/callbacklog", pwd)

	files, err := os.ReadDir(logPath)
	if err != nil {
		log.Printf("retryFailCallbackFile open dir err=%s", err.Error())
	}

	for _, v := range files {
		if !v.IsDir() {
			file, err := os.Open(v.Name())
			if err != nil {
				log.Printf("retryFailCallbackFile open file filename=%s err=%s", v.Name(), err.Error())
				continue
			}
			defer file.Close()

			d := gob.NewDecoder(file)
			var params []*biz.HandleCallbackParam
			err = d.Decode(&params)
			if err != nil {
				log.Printf("retryFailCallbackFile decode file filename=%s err=%s", v.Name(), err.Error())
			}

			os.Remove(v.Name())

			t.doCallback(params)
		}
	}
}
